package udf.string;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

public class NormalizeAgeType  extends EvalFunc<Tuple> {

	public Tuple exec(Tuple input) {
	
			if (input.size() != 1 ) {
				return null;
			}
	
			else {
	
				try {
					Tuple output = TupleFactory.getInstance().newTuple(1);
	
					String t = (String) input.get(0);
					t= t.toLowerCase().trim();
					String type = normalizeAgeType(t);
	
					//	System.out.println("Processing:" + domain + "\t" + level);
					output.set(0, type);
					return output;
	
				} catch (Exception e) {
					System.err.println("Failed to process input; error - "
							+ e.getMessage());
					return null;
				}
	
			}
		}

	
	/**
	 * years 6_7
	 *  years 8_9			kids
	 *  years_10_12
	 *  
	 *  
	 *  years_13_15			teens
	 *  
	 *  years_16_19			mteens
	 * 
	 * 
	 * kids-teens, kids-teens-mteens   
	 * 
	 *  
	 * 
	 * @param type
	 * @return
	 * 
	 *   51840 adults_25
 104911 adults_30
 289585 adults_40
 598068 adults_70
  14418 years_10_12
  40767 years_13_15
 170033 years_16_19
     90 years_4_5
    232 years_6_7
   2632 years_8_9

	 * 
	 * 
	 * 
	 */
	
	private String normalizeAgeType(String type){
		
		String t = type;

		if (type.equals("years_6_7")) {

			t = "kids";
		}

		else if (type.equals("years_8_9")) {

			t = "kids";
		}

		else if (type.equals("years_10_12")) {

			t = "kids";
		}

		else if (type.equals("years_13_15")) {

			t = "teens";
		}

		else if (type.equals("years_16_19")) {

			t = "mteens";
		}
		
		else if(type.equals("adults_25")){
			
			t="adults";
		}
		
		else if(type.equals("adults_30")){
			
			t="adults";
		}
		
		else if(type.equals("adults_40")){
			
			t="adults";
		}
		
		
		else if(type.equals("adults_70")){
			
			t="adults";
		}
		
		return t;
	}
	
}
